package drds.plus.repository.berkeley.spi;

import com.sleepycat.je.*;
import com.sleepycat.je.rep.ReplicaWriteException;
import drds.plus.executor.ExecuteContext;
import drds.plus.executor.cursor.cursor.ISortingCursor;
import drds.plus.executor.cursor.cursor.impl.SortingCursor;
import drds.plus.executor.record_codec.RecordCodecFactory;
import drds.plus.executor.record_codec.record.Record;
import drds.plus.executor.repository.Repository;
import drds.plus.executor.table.AbstractTable;
import drds.plus.executor.transaction.Transaction;
import drds.plus.executor.utils.Utils;
import drds.plus.sql_process.abstract_syntax_tree.configuration.IndexMapping;
import drds.plus.sql_process.abstract_syntax_tree.configuration.TableMetaData;
import drds.plus.sql_process.abstract_syntax_tree.execute_plan.query.QueryWithIndex;
import lombok.Getter;
import lombok.Setter;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

public class Table extends AbstractTable {
    @Setter
    @Getter
    boolean temporaryTable = false;
    @Setter
    @Getter
    Map<String, Database> tableNameToDatabaseMap = new HashMap();
    @Setter
    @Getter
    Environment environment = null;
    @Setter
    @Getter
    Map<String, KeyValueRecordCodec> indexToKeyValueRecordCodecMap;
    @Setter
    @Getter
    DatabaseEntry emptyValueEntry = new DatabaseEntry();

    {
        emptyValueEntry.setData(new byte[1]);
    }

    public Table(TableMetaData tableMetaData, Repository repository) {
        super(tableMetaData, repository);
        indexToKeyValueRecordCodecMap = new HashMap<String, KeyValueRecordCodec>();
        //
        KeyValueRecordCodec primaryIndexKeyValueRecordCodec = null;
        IndexMapping primaryIndexMapping = getTableMetaData().getPrimaryKeyIndexMetaData();
        if (primaryIndexKeyValueRecordCodec == null) {
            primaryIndexKeyValueRecordCodec = new KeyValueRecordCodec();
            primaryIndexKeyValueRecordCodec.setKeyRecordCodec(RecordCodecFactory.newRecordCodec((primaryIndexMapping.getKeyColumnMetaDataList())));
            if (primaryIndexMapping.getValueColumnMetaDataList() != null) {
                primaryIndexKeyValueRecordCodec.setValueRecordCodec(RecordCodecFactory.newRecordCodec((primaryIndexMapping.getValueColumnMetaDataList())));
            }
        }
        indexToKeyValueRecordCodecMap.put(primaryIndexMapping.getIndexName(), primaryIndexKeyValueRecordCodec);
        //
        for (IndexMapping secondIndexMapping : getTableMetaData().getSecondaryIndexNameToIndexMetaDataMapValues()) {
            KeyValueRecordCodec keyValueRecordCodec = new KeyValueRecordCodec();
            keyValueRecordCodec.setKeyRecordCodec(RecordCodecFactory.newRecordCodec((secondIndexMapping.getKeyColumnMetaDataList())));
            if (secondIndexMapping.getValueColumnMetaDataList() != null) {
                keyValueRecordCodec.setValueRecordCodec(RecordCodecFactory.newRecordCodec((secondIndexMapping.getValueColumnMetaDataList())));
            }
            indexToKeyValueRecordCodecMap.put(secondIndexMapping.getIndexName(), keyValueRecordCodec);
        }

    }


    public Database getDatabase(String tableName) {
        Database database = tableNameToDatabaseMap.get(tableName);
        if (database == null) {
            synchronized (this) {
                database = tableNameToDatabaseMap.get(tableName);
                if (database == null) {
                    database = ((RepositoryImpl) this.repository).getDatabase(tableName, tableMetaData.isTemporaryTable(), tableMetaData.isSortedDuplicates());
                    tableNameToDatabaseMap.put(tableName, database);
                }

            }
        }
        return database;
    }

    public void close() {
        for (Entry<String, Database> entry : tableNameToDatabaseMap.entrySet()) {
            entry.getValue().close();
        }
    }

    /**
     * todo:触发更新二级索引
     */
    public void put(ExecuteContext executeContext, String databaseName, IndexMapping indexMapping, Record key, Record value) throws RuntimeException {
        DatabaseEntry keyEntry = new DatabaseEntry();
        DatabaseEntry valueEntry = new DatabaseEntry();
        keyEntry.setData(indexToKeyValueRecordCodecMap.get(indexMapping.getIndexName()).getKeyRecordCodec().encode(key));
        //
        if (value != null) {
            valueEntry.setData(indexToKeyValueRecordCodecMap.get(indexMapping.getIndexName()).getValueRecordCodec().encode(value));
        } else {
            valueEntry = emptyValueEntry;
        }
        try {
            Transaction transaction = executeContext.getTransaction();
            com.sleepycat.je.Transaction $transaction = null;
            if (transaction != null && transaction instanceof TransactionImpl) {
                $transaction = ((TransactionImpl) transaction).transaction;
            }
            OperationStatus operationStatus = getDatabase(databaseName).put($transaction, keyEntry, valueEntry);
            if (operationStatus.equals(OperationStatus.SUCCESS)) {
                return;
            }
        } catch (LockTimeoutException ex) {
            throw ex;
        } catch (ReplicaWriteException ex) {
            throw new RuntimeException(ex);
        }
    }


    public void delete(ExecuteContext context, String databaseName, IndexMapping indexMapping, Record key) throws RuntimeException {
        DatabaseEntry keyEntry = new DatabaseEntry();
        keyEntry.setData(indexToKeyValueRecordCodecMap.get(indexMapping.getIndexName()).getKeyRecordCodec().encode(key));
        try {
            getDatabase(databaseName).delete(context.getTransaction() == null ? null : ((TransactionImpl) context.getTransaction()).transaction, keyEntry);
        } catch (LockTimeoutException ex) {
            throw ex;
        } catch (ReplicaWriteException ex) {
            throw new RuntimeException(ex);
        }
    }

    public Record get(ExecuteContext executeContext, String databaseName, IndexMapping indexMapping, Record key) {
        DatabaseEntry keyEntry = new DatabaseEntry();
        DatabaseEntry valueEntry = new DatabaseEntry();
        keyEntry.setData(indexToKeyValueRecordCodecMap.get(indexMapping.getIndexName()).getKeyRecordCodec().encode(key));
        OperationStatus operationStatus = getDatabase(databaseName).get(executeContext.getTransaction() == null ? null : ((TransactionImpl) executeContext.getTransaction()).transaction, keyEntry, valueEntry, LockMode.DEFAULT);
        if (OperationStatus.SUCCESS != operationStatus) {
            return null;
        }
        if (valueEntry.getSize() != 0) {
            return indexToKeyValueRecordCodecMap.get(indexMapping.getIndexName()).getValueRecordCodec().decode(valueEntry.getData());
        } else {
            return null;
        }
    }

    public ISortingCursor getCursor(ExecuteContext executeContext, QueryWithIndex queryWithIndex, IndexMapping indexMapping) throws RuntimeException {
        String tableName = queryWithIndex.getTableName();
        return getCursor(executeContext.getTransaction(), tableName, indexMapping);
    }

    public ISortingCursor getCursor(Transaction transaction, String tableName, IndexMapping indexMapping) throws RuntimeException {
        Database database = getDatabase(tableName);
        if (database == null) {
            throw new IllegalArgumentException("where don't contains indexName:" + tableName);
        }
        CursorConfig cursorConfig = CursorConfig.READ_COMMITTED;
        LockMode lockMode = LockMode.READ_UNCOMMITTED;
        CursorImpl cursor = new CursorImpl(indexMapping, database.openCursor(transaction == null ? null : ((TransactionImpl) transaction).transaction, cursorConfig), lockMode);
        if (transaction != null) {
            ((TransactionImpl) transaction).addCursor(cursor);
        }
        return new SortingCursor(cursor, cursor.getCursorMetaData(), Utils.getOrderByList(indexMapping));
    }

    public ISortingCursor getCursor(ExecuteContext executeContext, String actualTableName, IndexMapping indexMapping) throws RuntimeException {
        Database database = getDatabase(actualTableName);
        if (database == null) {
            throw new IllegalArgumentException("where don't contains indexName:" + actualTableName);
        }
        Transaction transaction = executeContext.getTransaction();
        CursorConfig cursorConfig = CursorConfig.DEFAULT;
        LockMode lockMode = LockMode.DEFAULT;
        if (transaction != null) {
            com.sleepycat.je.TransactionConfig transactionConfig = ((TransactionImpl) transaction).transactionConfig;
            if (transactionConfig.getReadUncommitted()) {
                cursorConfig = CursorConfig.READ_UNCOMMITTED;
                lockMode = LockMode.READ_UNCOMMITTED;
            } else if (transactionConfig.getReadCommitted()) {
                cursorConfig = CursorConfig.READ_COMMITTED;
                lockMode = LockMode.READ_COMMITTED;
            }
        } else {
            cursorConfig = CursorConfig.READ_COMMITTED;
        }
        CursorImpl cursor = new CursorImpl(indexMapping, database.openCursor(transaction == null ? null : ((TransactionImpl) transaction).transaction, cursorConfig), lockMode);
        if (transaction != null) {
            ((TransactionImpl) transaction).addCursor(cursor);
        }
        return new SortingCursor(cursor, cursor.getCursorMetaData(), Utils.getOrderByList(indexMapping));
    }
}
